package com.ruoyi.management.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.management.base.config.AmmeterMessageConstants;
import com.ruoyi.management.base.utils.Crc16Utils;
import com.ruoyi.management.base.utils.HexUtils;
import com.ruoyi.management.dao.mapper.AmmeterMapper;
import com.ruoyi.management.mqtt.MqttContext;
import com.ruoyi.management.mqtt.MqttProducer;
import com.ruoyi.management.pojo.entity.OperationInstructions;
import com.ruoyi.management.pojo.param.TidParam;
import com.ruoyi.management.pojo.po.AmmeterResultPo;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;


import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

@Slf4j
@Service
public class AmmeterTaskImpl {

    @Autowired
    private MqttProducer mqttProducer;

    @Autowired
    private AmmeterMapper ammeterMapper;

    @Autowired
    private MqttContext mqttContext;

    @Autowired
    @Qualifier("managementRedisTemplate")
    private RedisTemplate<String, AmmeterResultPo> redisTemplate;

    /**
     * 项目启动时立即执行
     */
    @PostConstruct
    public void init() throws Exception {
//        selectAllAmmeter();
    }

    /**
     * 定时查询所有电表指令
     * 每日5-7点各执行一次
     */
    @Scheduled(cron = "0 0 5-7 * * ?")
    public synchronized void selectAllAmmeter() {
        log.info("=====开始查询所有电表指令=====");

        // 查询所有智联器终端信息
        List<TidParam> tidList = ammeterMapper.selectTid();
        if (CollectionUtils.isEmpty(tidList)) {
            log.warn("未查询到任何电表信息");
            return;
        }

        // 1. 使用Java Stream按ptopic分组
        Map<String, List<TidParam>> groupedByTopic = tidList.stream()
                .collect(Collectors.groupingBy(TidParam::getPtopic));

        // 2. 创建线程池（推荐使用Spring管理的线程池）
        ExecutorService executorService = Executors.newCachedThreadPool();

        // 3. 并发执行每个topic下的所有电表指令发送
        groupedByTopic.forEach((topic, params) -> {
            executorService.submit(() -> {
                log.info("开始处理 topic: {}", topic);
                for (TidParam param : params) {
                    Integer tid = param.getTid(); // 操作地址id
                    if (param.getZt().equals("江环项目")) {
                        selectDirectives(tid, topic, param.getZlid());
                    }
                    if (param.getZt().equals("黄埭项目")) {
                        selectDirectivesHD(tid, topic, param.getZlid(), String.valueOf(param.getModelId()));
                    }


                }
            });
        });

        // 4. 关闭线程池（注意：实际生产中应复用线程池，而不是每次都创建关闭）
        executorService.shutdown();
    }


    /**
     * 发送单独电表所有指令方法
     *
     * @param tid   电表ID
     * @param topic 智联器主题
     * @param zlid  智联器终端ID
     */
    public void selectDirectives(Integer tid, String topic, String zlid) {

        //存入操作数据库中
        OperationInstructions operationInstructions = new OperationInstructions();
        operationInstructions.setDbid(String.valueOf(tid)); // 电表ID
        operationInstructions.setZlid(zlid); // 智联器终端ID
        String ammeterId = Integer.toHexString(tid);
        for (int directives = 1; directives <= 1; directives++) {
            if (directives == 1) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, directivesHex,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        AmmeterMessageConstants.AMMETER_PUBLISH_TWO);
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 2) {
                continue;
            }
            String directivesHex = Integer.toHexString(directives);
            //数据入库
            operationInstructions.setDirectives(directivesHex);
            operationInstructions.setCreatedTime(new java.util.Date());

            ammeterMapper.insert(operationInstructions);
            log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

            // 计算CRC校验码
            byte[] crcHexString = Crc16Utils.getData(ammeterId,
                    AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                    AmmeterMessageConstants.AMMETER_PUBLISH_NULL, directivesHex,
                    AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                    AmmeterMessageConstants.AMMETER_PUBLISH_NUll);
            // 将CRC校验码转换为十六进制字符串并转换为大写
            String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
            try {
                byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                // 发送MQTT消息
                mqttProducer.send(topic, payload);
                log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                Thread.sleep(3000);
            } catch (Exception e) {
                // 捕获并抛出异常
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 黄埭项目指令发送
     *
     * @param tid    电表地址ID
     * @param topic  mqtt topic
     * @param zlid   智联器终端ID
     * @param params 参数 0-电表类型
     */
    public void selectDirectivesHD(Integer tid, String topic, String zlid, String... params) {
        //存入操作数据库中
        OperationInstructions operationInstructions = new OperationInstructions();
        operationInstructions.setDbid(String.valueOf(tid)); // 电表ID
        operationInstructions.setZlid(zlid); // 智联器终端ID
        String ammeterId = Integer.toHexString(tid);
        for (int directives = 1; directives <= 1; directives++) {
            /* 总电量查询 */
            if (directives == 1) {
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 发送指令
                String finalMessageHex = null;

                /* 安科瑞ADL400 */
                if (ObjectUtils.isNotEmpty(params) && 1 == NumberUtils.toLong(params[0], -1)) {
                    // 计算CRC校验码
                    byte[] crcHexString = Crc16Utils.getData(ammeterId,
                            AmmeterMessageConstants.AMMETER_PUBLISH_READ3,
                            AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                            AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                            AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                            AmmeterMessageConstants.AMMETER_PUBLISH_TWO);
                    // 将CRC校验码转换为十六进制字符串并转换为大写
                    finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                }
                /* 安科瑞527AMC */
                else if (StringUtils.isNotEmpty(params) && 2 == NumberUtils.toLong(params[0], -1)) {
                    // 计算CRC校验码
                    byte[] crcHexString = Crc16Utils.getData(ammeterId,
                            AmmeterMessageConstants.AMMETER_PUBLISH_READ3,
                            AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                            AmmeterMessageConstants.AMMETER_PUBLISH_READ47,
                            AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                            AmmeterMessageConstants.AMMETER_PUBLISH_TWO);
                    // 将CRC校验码转换为十六进制字符串并转换为大写
                    finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                }
                /* 海恩德 */
                else if (StringUtils.isNotEmpty(params) && 3 == NumberUtils.toLong(params[0], -1)) {
                    // 计算CRC校验码
                    byte[] crcHexString = Crc16Utils.getData(ammeterId,
                            AmmeterMessageConstants.AMMETER_PUBLISH_READ3,
                            AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                            AmmeterMessageConstants.AMMETER_PUBLISH_READ47,
                            AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                            AmmeterMessageConstants.AMMETER_PUBLISH_THREE);
                    // 将CRC校验码转换为十六进制字符串并转换为大写
                    finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                } else {
                    // 计算CRC校验码
                    byte[] crcHexString = Crc16Utils.getData(ammeterId,
                            AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                            AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "1D",
                            AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                            AmmeterMessageConstants.AMMETER_PUBLISH_TWO);
                    // 将CRC校验码转换为十六进制字符串并转换为大写
                    finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                }
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 2) {
                continue;
            }
            if (directives == 3) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "08",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 4) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "00",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 5) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "01",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 6) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "02",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 7) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "03",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 8) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "04",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 9) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "05",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 10) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "21",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 11) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "22",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 12) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "23",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 13) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "1A",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 14) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "1B",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 15) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "1C",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
            if (directives == 16) {
                //总电量查询
                String directivesHex = Integer.toHexString(directives);
                //数据入库
                operationInstructions.setDirectives(directivesHex);
                operationInstructions.setCreatedTime(new java.util.Date());
                ammeterMapper.insert(operationInstructions);
                log.info("插入操作指令: {}, 智联器地址: {}, 电表地址: {}, ", directives, zlid, tid);

                // 计算CRC校验码
                byte[] crcHexString = Crc16Utils.getData(ammeterId,
                        AmmeterMessageConstants.AMMETER_PUBLISH_READ4,
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL, "12",
                        AmmeterMessageConstants.AMMETER_PUBLISH_NULL,
                        "01");
                // 将CRC校验码转换为十六进制字符串并转换为大写
                String finalMessageHex = Crc16Utils.byteTo16String(crcHexString).toUpperCase();
                try {
                    byte[] payload = HexUtils.HexStringToByte(finalMessageHex);
                    // 发送MQTT消息
                    mqttProducer.send(topic, payload);
                    log.info("发送MQTT消息: 操作指令: {}, 智联器地址: {}, 电表地址: {}, 最终消息（十六进制）: {}", directives, zlid, tid, finalMessageHex);
                    Thread.sleep(3000);
                } catch (Exception e) {
                    // 捕获并抛出异常
                    throw new RuntimeException(e);
                }
                continue;
            }
        }
    }

    /**
     * 清理操作日志
     * 每日20:30执行一次
     */
    @Scheduled(cron = "0 30 20 * * ?")
    public synchronized void checkAndCleanDatabase() {
        log.info("开始检查数据库...");
        int currentCount = mqttContext.getAmmeterResultMapper().countAll();
        if (currentCount > 100) {
            int recordsToDelete = currentCount - 100;
            mqttContext.getAmmeterResultMapper().deleteOldestRecords(recordsToDelete);
            log.info("删除了 {} 条旧数据，当前记录数: {}", recordsToDelete, 100);
        } else {
            log.info("当前记录数: {}", currentCount);
        }
    }


    /**
     * 存入每天最后电表记录
     * 每日08:30执行一次
     */
    @Scheduled(cron = "0 30 8 * * ?")
    public void saveEveryDataEachMonth() {
        log.info("=====开始：保存每个电表最接近当前时间的数据到数据库=====");
        long startTime = System.currentTimeMillis();

        // 扫描 Redis 数据
        ScanOptions options = ScanOptions.scanOptions().match("ammeter:result:*").build();
        Cursor<byte[]> scan = mqttContext.getRedisTemplate().getConnectionFactory().getConnection().scan(options);

        // 创建一个 Map，用于按电表编号分组
        Map<String, List<AmmeterResultPo>> groupedByDbbh = new HashMap<>();

        while (scan.hasNext()) {
            byte[] keyBytes = scan.next();
            String redisKey = new String(keyBytes);
            AmmeterResultPo resultPo = mqttContext.getRedisTemplate().opsForValue().get(redisKey);

            if (resultPo != null) {
                String dbbh = resultPo.getDbbh();
                // 根据电表编号分组将数据添加到集合中
                groupedByDbbh.computeIfAbsent(dbbh, k -> new ArrayList<>()).add(resultPo);
            }
        }

        // 获取当前时间
        long now = System.currentTimeMillis();

        // 遍历分组
        for (Map.Entry<String, List<AmmeterResultPo>> entry : groupedByDbbh.entrySet()) {
            String dbbh = entry.getKey();
            List<AmmeterResultPo> dataList = entry.getValue();


            // 1. 获取数据库中最新的 totalpower（从 dbday 表）
            BigDecimal latestDbPower = ammeterMapper.getLatestTotalPowerByDbbh(dbbh);

            // 2. 按时间排序 排序顺序： 按照时间升序
            dataList.sort(Comparator.comparing(AmmeterResultPo::getTimestamp));

            // 3. 遍历数据，找到第一条符合条件（totalpower >= 数据库值）的记录
            AmmeterResultPo selectedRecord = null;
            for (int i = dataList.size() - 1; i >= 0; i--) {
                AmmeterResultPo currentRecord = dataList.get(i);
                if (latestDbPower == null || currentRecord.getTotalPower().compareTo(latestDbPower) >= 0) {
                    selectedRecord = currentRecord;
                    break;
                }
            }

            // 4. 如果没有符合条件的记录，则取时间最接近的一条（兜底逻辑）
            if (selectedRecord == null) {
                selectedRecord = dataList.stream()
                        .min(Comparator.comparingLong(po -> Math.abs(po.getTimestamp().getTime() - now)))
                        .orElse(null);
            }

            // 5. 存入数据库
            if (selectedRecord != null) {
                try {
                    Boolean success = ammeterMapper.insertDay(selectedRecord);
                    if (success) {
                        log.info("已插入电表 {} 的数据, 时间：{}, total power：{}",
                                selectedRecord.getDbbh(),
                                selectedRecord.getTimestamp(),
                                selectedRecord.getTotalPower());
                    } else {
                        log.info("插入电表数据失败：{}", selectedRecord.getDbbh());
                    }
                } catch (Exception e) {
                    log.error("插入电表数据失败：{}", selectedRecord.getDbbh(), e);
                }
            }
        }

        log.info("=====完成：Redis 数据已处理并存入数据库，耗时 {} ms =====", System.currentTimeMillis() - startTime);
    }


    /**
     * 存入月度数据
     * 每月最后一天08:30执行一次
     */
    @Scheduled(cron = "0 30 8 L * ?")
    public void saveFirstDataEachMonth() {
        log.info("开始：保存每个电表最接近当前时间的数据到数据库");

        long startTime = System.currentTimeMillis();

        ScanOptions options = ScanOptions.scanOptions().match("ammeter:result:*").build();
        Cursor<byte[]> scan = mqttContext.getRedisTemplate().getConnectionFactory().getConnection().scan(options);

        Map<String, List<AmmeterResultPo>> groupedByDbbh = new HashMap<>();

        while (scan.hasNext()) {
            byte[] keyBytes = scan.next();
            String redisKey = new String(keyBytes);
            AmmeterResultPo resultPo = mqttContext.getRedisTemplate().opsForValue().get(redisKey);

            if (resultPo != null) {
                String dbbh = resultPo.getDbbh();
                groupedByDbbh.computeIfAbsent(dbbh, k -> new ArrayList<>()).add(resultPo);
            }
        }

        // 获取当前时间作为参考点
        long now = System.currentTimeMillis();

        // 处理每个电表的数据
        for (Map.Entry<String, List<AmmeterResultPo>> entry : groupedByDbbh.entrySet()) {
            String dbbh = entry.getKey();
            List<AmmeterResultPo> dataList = entry.getValue();

            // 取时间上最接近 now 的那条数据
            Optional<AmmeterResultPo> closest = dataList.stream()
                    .min(Comparator.comparingLong(po -> Math.abs(po.getTimestamp().getTime() - now)));

            // 存入数据库
            closest.ifPresent(po -> {
                try {
                    Boolean a = ammeterMapper.insertmonth(po);
                    if (a) {
                        log.info("已插入电表 {} 的最接近时间数据，时间：{}", po.getDbbh(), po.getTimestamp());
                    } else {
                        log.info("插入电表数据失败：{}", po.getDbbh());
                    }
                } catch (Exception e) {
                    log.error("插入电表数据失败：{}", po.getDbbh(), e);
                }
            });
        }
        log.info("完成：Redis 数据已处理并存入数据库，耗时 {} ms", System.currentTimeMillis() - startTime);
    }

    /**
     * 存入年度数据
     * 每年最后一天08:30执行一次
     */
    @Scheduled(cron = "0 30 8 31 12 ?")
    public void saveFirstDataEachYear() {
        log.info("开始：保存每个电表最接近当前时间的数据到数据库");

        long startTime = System.currentTimeMillis();

        ScanOptions options = ScanOptions.scanOptions().match("ammeter:result:*").build();
        Cursor<byte[]> scan = mqttContext.getRedisTemplate().getConnectionFactory().getConnection().scan(options);

        Map<String, List<AmmeterResultPo>> groupedByDbbh = new HashMap<>();

        while (scan.hasNext()) {
            byte[] keyBytes = scan.next();
            String redisKey = new String(keyBytes);
            AmmeterResultPo resultPo = mqttContext.getRedisTemplate().opsForValue().get(redisKey);

            if (resultPo != null) {
                String dbbh = resultPo.getDbbh();
                groupedByDbbh.computeIfAbsent(dbbh, k -> new ArrayList<>()).add(resultPo);
            }
        }

        // 获取当前时间作为参考点
        long now = System.currentTimeMillis();
        // 处理每个电表的数据
        for (Map.Entry<String, List<AmmeterResultPo>> entry : groupedByDbbh.entrySet()) {
            String dbbh = entry.getKey();
            List<AmmeterResultPo> dataList = entry.getValue();

            // 取时间上最接近 now 的那条数据
            Optional<AmmeterResultPo> closest = dataList.stream()
                    .min(Comparator.comparingLong(po -> Math.abs(po.getTimestamp().getTime() - now)));

            // 存入数据库
            closest.ifPresent(po -> {
                try {
                    Boolean a = ammeterMapper.insertYear(po);
                    if (a) {
                        log.info("已插入电表 {} 的最接近时间数据，时间：{}", po.getDbbh(), po.getTimestamp());
                    } else {
                        log.info("插入电表数据失败：{}", po.getDbbh());
                    }
                } catch (Exception e) {
                    log.error("插入电表数据失败：{}", po.getDbbh(), e);
                }

            });
        }
        log.info("完成：Redis 数据已处理并存入数据库，耗时 {} ms", System.currentTimeMillis() - startTime);
    }
}

